package batch;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;

public class operator_outjoin {
    public static void main(String[] args) throws Exception {
    //    创建运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    //    获取数据
        ArrayList<Tuple2<Integer,String>> data1 = new ArrayList<Tuple2<Integer,String>>();
        data1.add(new Tuple2<Integer, String>(1,"bing"));
        data1.add(new Tuple2<Integer, String>(2,"shui"));
        data1.add(new Tuple2<Integer, String>(3,"fan"));


        ArrayList<Tuple2<Integer,String>> data2 = new ArrayList<Tuple2<Integer,String>>();
        data2.add(new Tuple2<Integer, String>(1,"beijin"));
        data2.add(new Tuple2<Integer, String>(3,"wuhan"));
        data2.add(new Tuple2<Integer, String>(4,"gz"));


        DataSet<Tuple2<Integer,String>> user =  env.fromCollection(data1);
        DataSet<Tuple2<Integer,String>> city =  env.fromCollection(data2);

        System.out.println("*****leftoutjoin****");
        user.leftOuterJoin(city).
                where(0).
                equalTo(0).
                with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String,String>>() {
            public Tuple3<Integer, String,String> join(Tuple2<Integer, String> left_, Tuple2<Integer, String> right_) throws Exception {
                if (right_ == null ){
                    return new Tuple3<Integer,String,String>(left_.f0,left_.f1,null);
                }
                else{
                    return new Tuple3<Integer,String,String>(left_.f0,left_.f1,right_.f1);
                }
            }
        }).print();



        System.out.println("*****rightoutjoin****");
        user.rightOuterJoin(city).
                where(0).
                equalTo(0).
                with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String,String>>() {
                    public Tuple3<Integer, String,String> join(Tuple2<Integer, String> left_, Tuple2<Integer, String> right_) throws Exception {
                        if (left_ == null ){
                            return new Tuple3<Integer,String,String>(right_.f0,right_.f1,null);
                        }
                        else{
                            return new Tuple3<Integer,String,String>(right_.f0,right_.f1,left_.f1);
                        }
                    }
                }).print();

        System.out.println("*****fulloutjoin****");
        user.fullOuterJoin(city).
                where(0).
                equalTo(0).
                with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Tuple3<Integer, String,String>>() {
                    public Tuple3<Integer, String,String> join(Tuple2<Integer, String> left_, Tuple2<Integer, String> right_) throws Exception {
                        if (left_ == null ){
                            return new Tuple3<Integer,String,String>(right_.f0,right_.f1,null);
                        }else if(right_ == null){
                            return new Tuple3<Integer,String,String>(left_.f0,null,left_.f1);
                        }
                        else{
                            return new Tuple3<Integer,String,String>(right_.f0,right_.f1,left_.f1);
                        }
                    }
                }).print();

    }
}
